package com.mm.mindmap.consistency.strategy.scheme;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.mm.common.utils.theadPoolUtil.*;
import com.mm.common.utils.theadPoolUtil.strategy.error.RetryError;
import com.mm.mindmap.consistency.ConfigVar;
import com.mm.mindmap.consistency.ConsistencyException;
import com.mm.mindmap.consistency.ConsistencyTask;
import com.mm.mindmap.consistency.ConsistencyData;
import com.mm.mindmap.consistency.dataFunc.DataFunc;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;


/**
 * @Program: 8.mindmap
 * @Author: 海里的鱼
 * @Create: 2024-08-28 09:27
 * @Name: 【MQ】方案
 * @Desc: 使用消息队列来实现数据一致性的延迟更新
 */

@Component
@RequiredArgsConstructor
public class MQScheme implements IConsistencyScheme{
    //RabbitMq的客户端
    private final RabbitTemplate rabbitTemplate;
    //Spring容器
    private final ApplicationContext context;
    //消费批量处理器，用于批量处理消息的消费（线程池），设置重试机制（3次，间隔1000毫秒）
    private BulkExecuteHandler consumeHandler = BulkExecuteUtil.getHandler(new RetryError(ConfigVar.MAX_RETRY_MQCONSUSME,ConfigVar.INTERVAL_MQCONSUSME));


    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 静态内部类MQData
     * @Describe: 包装【ConsistencyData】的data，,data.class和dataFunc以发送至RabbitMQ
     **/
     static class MQData{
        public Object data;
        public Class<ConsistencyData> writeFunc;
        public Class dataClass;

        public MQData(Object data, Class writeFunc, Class dataClass) {
            this.data = data;
            this.writeFunc = writeFunc;
            this.dataClass = dataClass;
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 执行方案
     * @Describe: 先发送消息至消息队列，再由消费者消费
     **/
     @Override
    public boolean executeScheme(ConsistencyData data) {
         MQData mqData = new MQData(data.getData(),data.getWriteFunc().getClass(),data.getData().getClass());
         rabbitTemplate.convertAndSend(data.getExchangeName(), data.getRoutingKey(),JSONUtil.toJsonStr(mqData));
        return true;
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 消费消息
     * @Describe: TODO
     **/
    private boolean consume(String json){
        try {
            JSONObject jsonObject = JSONUtil.parseObj(json);
            Class dataClass = Class.forName(jsonObject.getStr("dataClass").replaceFirst("^class ", ""));
            Class<DataFunc> writeFunc = (Class<DataFunc>) Class.forName(jsonObject.getStr("writeFunc").replaceFirst("^class ", ""));

            Object data = JSONUtil.toBean(jsonObject.getStr("data"),dataClass);
            ConsistencyData consistencyData = new ConsistencyData(data, null, context.getBean(writeFunc), null);
            ConsistencyTask consistencyTask = new ConsistencyTask(consistencyData);
            return consumeHandler.executeTask(consistencyTask);

        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    /** --------------------------------------------------------------------------------------------------------------------
     * @Method  : 监听队列
     * @Describe: TODO
     **/
     @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "mysql.send.queue"),
            exchange = @Exchange(name = "consistency.direct"),
            key = "mysql.send"
    ))
    public void listenMysql(String json){
         if(!consume(json)){
             new ConsistencyException("【MQConsistency】【listenMysql】 消费MySQL消息失败").printStackTrace();
         }
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "es.send.queue"),
            exchange = @Exchange(name = "consistency.direct"),
            key = "es.send"
    ))
    public void listenES(String json){
        if(!consume(json)){
            new ConsistencyException("【MQConsistency】【listenES】 消费ES消息失败").printStackTrace();
        }
    }


    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "redis.send.queue"),
            exchange = @Exchange(name = "consistency.direct"),
            key = "redis.send"
    ))
    public void listenRedis(String json){
        if(!consume(json)){
            new ConsistencyException("【MQConsistency】【listenRedis】 消费Redis消息失败").printStackTrace();
        }
    }



}
